/*
 * Copyright (C) 2020-present, kinbug All rights reserved.
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.gitee.kinbug.sniper.web.websocket.handler;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.socket.HandshakeInfo;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;

import com.gitee.kinbug.sniper.web.websocket.annotations.WebSocketMapping;
import com.gitee.kinbug.sniper.web.websocket.entity.WebSocketSender;
import com.gitee.kinbug.sniper.web.websocket.msg.SymbolMsg;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * -ws 链接
 * @ClassName: EchoHandler
 * @author kinbug
 * @date 2020年9月14日
 */
@Component
@WebSocketMapping("/subSymbol")
public class EchoHandler implements WebSocketHandler {

	@Autowired
	private ConcurrentHashMap<String, WebSocketSender> senderMap;
	@Autowired
	private SymbolMsg symbolMsg;
	
	@Override
	public Mono<Void> handle(WebSocketSession session) {
		HandshakeInfo handshakeInfo = session.getHandshakeInfo();
		Map<String, String> queryMap = getQueryMap(handshakeInfo.getUri().getQuery());
		String id = queryMap.getOrDefault("id", "defaultId");
		Mono<Void> input = session.receive().map(WebSocketMessage::getPayloadAsText)
				.doOnNext(obj -> symbolMsg.msgReceive(obj)).then();
		Mono<Void> output = session.send(Flux.create(sink -> senderMap.put(id, new WebSocketSender(session, sink))));
		/**
		 * Mono.zip() 会将多个 Mono 合并为一个新的 Mono，任何一个 Mono 产生 error 或 complete 都会导致合并后的 Mono
		 * 也随之产生 error 或 complete，此时其它的 Mono 则会被执行取消操作。
		 */
		return Mono.zip(input, output).then();
	}

	/**
	 * -用于获取url参数 
	 */ 
	private Map<String, String> getQueryMap(String queryStr) {
		Map<String, String> queryMap = new HashMap<>(16);
		if (!StringUtils.isEmpty(queryStr)) {
			String[] queryParam = queryStr.split("&");
			Arrays.stream(queryParam).forEach(s -> {
				String[] kv = s.split("=", 2);
				String value = kv.length == 2 ? kv[1] : "";
				queryMap.put(kv[0], value);
			});
		}
		return queryMap;
	}
}
